package com.hgl.bi.web.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hgl.bi.common.config.COSClientConfig;
import com.hgl.bi.common.constant.ErrorCode;
import com.hgl.bi.common.constant.ProgressStatus;
import com.hgl.bi.common.exception.BusinessException;
import com.hgl.bi.common.manager.COSManager;
import com.hgl.bi.common.utils.AssertUtil;
import com.hgl.bi.common.utils.FileUtil;
import com.hgl.bi.web.config.DeepSeekClient;
import com.hgl.bi.web.config.RabbitConfig;
import com.hgl.bi.web.entity.CleanEntity;
import com.hgl.bi.web.entity.MetadataEntity;
import com.hgl.bi.web.model.dto.CleanMsgDto;
import com.hgl.bi.web.service.AnalyzeService;
import com.hgl.bi.web.service.CleanService;
import com.hgl.bi.web.service.MetadataService;
import com.hgl.bi.web.service.UserService;
import com.qcloud.cos.COSClient;
import com.qcloud.cos.model.PutObjectResult;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @program: BI
 * @description: 数据清洗消费者
 * @author: hgl
 * @create: 2025-02-15 23:17
 */
@Slf4j
@Component
public class CleanConsumer {

    @Autowired
    private CleanService cleanService;

    @Autowired
    private AnalyzeService analyzeService;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private DeepSeekClient deepSeekClient;

    @Autowired
    private COSManager cosManager;

    @Autowired
    private MetadataService metadataService;

    @Autowired
    private COSClient cosClient;

    @Autowired
    private COSClientConfig cosClientConfig;

    @Autowired
    private UserService userService;

    @RabbitListener(queues = RabbitConfig.DATA_CLEAN_QUEUE)
    public void doDataClean(@Payload CleanMsgDto messageDto, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        // 解析消息
        Long cleanId = messageDto.getCleanId();
        Long fileId = messageDto.getFileId();
        // 非空校验
        AssertUtil.notNull(fileId, ErrorCode.FILE_ID_NULL);
        AssertUtil.notNull(cleanId, ErrorCode.CLEAN_ID_NULL);

        try {
            // 1. 获取待处理任务（使用乐观锁）
            CleanEntity clean = cleanService.lambdaQuery()
                    .eq(CleanEntity::getId, cleanId)
                    .eq(CleanEntity::getProgress, ProgressStatus.WAITING.getCode())
                    .oneOpt()
                    .orElseThrow(() -> new BusinessException(ErrorCode.TASK_NOT_EXIST_OR_PROCESSED));
            // 2. 更新为处理中状态
            clean.setProgress(ProgressStatus.RUNNING.getCode());
            clean.setStartTime(new Date());
            if (!cleanService.updateById(clean)) {
                throw new BusinessException(ErrorCode.STATUS_UPDATE_ERROR);
            }
            // 3. 执行业务逻辑
            processCleaning(clean, fileId);
            // 4. 确认消息
            channel.basicAck(tag, false);
        } catch (BusinessException e) {
            // 业务异常不重试
            log.error("任务处理失败 taskId:{}", cleanId, e);
            basicNackWithRetry(channel, tag); // 最大重试3次
        } catch (Exception e) {
            log.error("系统异常 taskId:{}", cleanId, e);
            basicNackWithRetry(channel, tag);
        }
    }

    private void processCleaning(CleanEntity clean, Long fileId) {
        try {
            // 获取cos path
            MetadataEntity metadata = metadataService.getById(fileId);
            String cosPath = metadata.getCosPath();
            // 读取文件内容
            List<Map<Integer, String>> content = cosManager.syncReadContent(cosPath);
            // 转换为CSV格式
            String csv = FileUtil.toCsv(content);
            log.info("CSV数据: {}", csv);
            // 调用DeepSeek API
            String prompt = FileUtil.loadPromptText(true);
            log.info("提示词: {}", prompt);
            log.info("正在调用{}模型执行清洗任务: {}",deepSeekClient.getModel(), clean.getId());
            String result = deepSeekClient.callDeepSeekApi(prompt, "csv数据:\n" + csv);
            // 非空校验
            AssertUtil.notEmpty(result, ErrorCode.RESPONSE_IS_NULL);
            String[] split = result.split("【【【");
            AssertUtil.isTrue(!(split.length < 3), ErrorCode.RESPONSE_ERROR);
            String data = split[1].trim();
            String cleanLog = split[2].trim();
            String[] split1 = cleanLog.split("总金额：");
            String inconmeStr = split1[1].split("元")[0].trim();
            BigDecimal income = new BigDecimal(inconmeStr);

            // 更新任务状态
            clean.setEndTime(new Date());
            clean.setCleanLog(cleanLog);
            clean.setProgress(ProgressStatus.SUCCESS.getCode());
            // 保存结果
            File tempFile = null;
            String cosCsvPath = "";
            try {
                tempFile = FileUtil.createTempCSV(data);
                // 找到最后一个 "/" 的位置
                int lastSlashIndex = cosPath.lastIndexOf("/");

                // 提取从开头到最后一段路径的部分
                String prefix = cosPath.substring(0, lastSlashIndex + 1);
                cosCsvPath =  prefix + tempFile.getName();
                PutObjectResult putObjectResult = cosClient.putObject(cosClientConfig.getBucketName(), cosCsvPath, tempFile);

            } catch (IOException e) {
                throw new RuntimeException("文件处理失败", e);
            } finally {
                // 3. 清理临时文件
                if (tempFile != null && tempFile.exists()) {
                    tempFile.delete();
                }
            }
            metadata.setCosCsvPath(cosCsvPath);
            metadata.setCleanCount(metadata.getCleanCount() + 1);
            metadata.setIsCleaned(1);
            metadata.setIncome(income);
            metadataService.updateById(metadata);
            cleanService.updateById(clean);
        } catch (Exception e) {
            // 记录详细错误日志
            clean.setEndTime(new Date());
            clean.setErrorLog(ExceptionUtils.getMessage(e));
            clean.setProgress(ProgressStatus.FAILURE.getCode());
            cleanService.updateById(clean);
        }
    }

    private void basicNackWithRetry(Channel channel, long tag) {
        try {
            GetResponse response = channel.basicGet(RabbitConfig.DATA_CLEAN_QUEUE, false);
            if (response == null) {
                channel.basicNack(tag, false, false);
                return;
            }

            AMQP.BasicProperties props = response.getProps();
            Integer retryCount = (Integer) props.getHeaders().getOrDefault("x-retry-count", 0);

            if (retryCount < 3) {
                // 重新发布消息
                channel.basicPublish("", RabbitConfig.DATA_CLEAN_QUEUE,
                        new AMQP.BasicProperties.Builder()
                                .headers(createHeaders(retryCount))
                                .build(),
                        response.getBody());

                channel.basicAck(tag, false);
            } else {
                channel.basicNack(tag, false, false); // 进入死信队列
            }
        } catch (IOException ex) {
            log.error("消息NACK处理失败", ex);
        }
    }

    // Java 8兼容写法替代Map.of()
    private Map<String, Object> createHeaders(int retryCount) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("x-retry-count", retryCount + 1);
        return headers;
    }
}
